package com.electric.flinkcep.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

/**
 * 配置管理器
 * 负责加载和管理应用程序配置
 */
public class ConfigManager {
    
    private static final Logger logger = LoggerFactory.getLogger(ConfigManager.class);
    private static final String CONFIG_FILE = "application.properties";
    private static ConfigManager instance;
    private Properties properties;
    
    private ConfigManager() {
        loadProperties();
    }
    
    public static synchronized ConfigManager getInstance() {
        if (instance == null) {
            instance = new ConfigManager();
        }
        return instance;
    }
    
    private void loadProperties() {
        properties = new Properties();
        try (InputStream inputStream = getClass().getClassLoader().getResourceAsStream(CONFIG_FILE)) {
            if (inputStream != null) {
                properties.load(inputStream);
                logger.info("配置文件加载成功: {}", CONFIG_FILE);
            } else {
                logger.warn("配置文件未找到: {}, 使用默认配置", CONFIG_FILE);
                loadDefaultProperties();
            }
        } catch (IOException e) {
            logger.error("加载配置文件失败: {}", CONFIG_FILE, e);
            loadDefaultProperties();
        }
    }
    
    private void loadDefaultProperties() {
        // 默认配置
        properties.setProperty("kafka.bootstrap.servers", "localhost:9092");
        properties.setProperty("kafka.group.id", "flink-cep-consumer");
        properties.setProperty("kafka.topic", "input-events");
        properties.setProperty("kafka.auto.offset.reset", "latest");
        
        properties.setProperty("http.backend.url", "http://localhost:8080/api/events");
        properties.setProperty("http.timeout.connection", "5000");
        properties.setProperty("http.timeout.socket", "10000");
        
        properties.setProperty("flink.parallelism", "1");
        properties.setProperty("flink.checkpoint.interval", "60000");
        properties.setProperty("flink.checkpoint.mode", "EXACTLY_ONCE");
        
        properties.setProperty("cep.pattern.timeout", "30000");
        properties.setProperty("cep.pattern.name", "alert-pattern");
        
        properties.setProperty("logging.level", "INFO");
    }
    
    public String getProperty(String key) {
        return properties.getProperty(key);
    }
    
    public String getProperty(String key, String defaultValue) {
        return properties.getProperty(key, defaultValue);
    }
    
    public int getIntProperty(String key, int defaultValue) {
        String value = properties.getProperty(key);
        if (value != null) {
            try {
                return Integer.parseInt(value);
            } catch (NumberFormatException e) {
                logger.warn("配置项 {} 的值 {} 不是有效的整数，使用默认值 {}", key, value, defaultValue);
            }
        }
        return defaultValue;
    }
    
    public long getLongProperty(String key, long defaultValue) {
        String value = properties.getProperty(key);
        if (value != null) {
            try {
                return Long.parseLong(value);
            } catch (NumberFormatException e) {
                logger.warn("配置项 {} 的值 {} 不是有效的长整数，使用默认值 {}", key, value, defaultValue);
            }
        }
        return defaultValue;
    }
    
    public boolean getBooleanProperty(String key, boolean defaultValue) {
        String value = properties.getProperty(key);
        if (value != null) {
            return Boolean.parseBoolean(value);
        }
        return defaultValue;
    }
    
    // Kafka配置获取方法
    public String getKafkaBootstrapServers() {
        return getProperty("kafka.bootstrap.servers", "localhost:9092");
    }
    
    public String getKafkaGroupId() {
        return getProperty("kafka.group.id", "flink-cep-consumer");
    }
    
    public String getKafkaTopic() {
        return getProperty("kafka.topic", "input-events");
    }
    
    public String getKafkaAutoOffsetReset() {
        return getProperty("kafka.auto.offset.reset", "latest");
    }
    
    // HTTP配置获取方法
    public String getHttpBackendUrl() {
        return getProperty("http.backend.url", "http://localhost:8080/api/events");
    }
    
    public int getHttpConnectionTimeout() {
        return getIntProperty("http.timeout.connection", 5000);
    }
    
    public int getHttpSocketTimeout() {
        return getIntProperty("http.timeout.socket", 10000);
    }
    
    // Flink配置获取方法
    public int getFlinkParallelism() {
        return getIntProperty("flink.parallelism", 1);
    }
    
    public long getFlinkCheckpointInterval() {
        return getLongProperty("flink.checkpoint.interval", 60000L);
    }
    
    public String getFlinkCheckpointMode() {
        return getProperty("flink.checkpoint.mode", "EXACTLY_ONCE");
    }
    
    // CEP配置获取方法
    public long getCepPatternTimeout() {
        return getLongProperty("cep.pattern.timeout", 30000L);
    }
    
    public String getCepPatternName() {
        return getProperty("cep.pattern.name", "alert-pattern");
    }
}